package com.example.kafkaproducer.service.impl;

import com.example.kafkaproducer.service.SendService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.concurrent.TimeUnit;

/**
 * @author: wukangrong
 * @date: 2021/1/29 15:07
 * @description:
 */
@Component
@Slf4j
public class SendServiceImpl implements SendService, CommandLineRunner {


    @Autowired
    KafkaTemplate kafkaTemplate;

    @Value("${kafka.custom.topic.name}")
    private String topicName;

    @Override
    public void send() {
        String data = "hello kafka time:" + LocalDateTime.now().toString();
        kafkaTemplate.send(topicName, "wkr", data)
                .addCallback((msg) -> log.info("send success, response:{}", msg), (exception) -> log.error("send error, exceptio:{}", exception.getMessage(), exception));
        kafkaTemplate.send("name", "wu", data)
                .addCallback((msg) -> log.info("send2 success, response:{}", msg), (exception) -> log.error("send2 error, exceptio:{}", exception.getMessage(), exception));

    }

    @Override
    public void run(String... args) throws InterruptedException {
        log.info("准备发送数据");
        TimeUnit.SECONDS.sleep(2);
        send();
        log.info("发送数据完成");
    }
}
